package com.lg.customrxjava;

public class ObservableMap<T,R> implements ObservableOnSubscribe<R> {

    ObservableOnSubscribe<T> source;

    Function<T,R> function;

    Observer<R> observableEmitter;

    public ObservableMap(ObservableOnSubscribe source, Function<T,R> function) {
        this.source = source;
        this.function = function;
    }



    @Override
    public void subscribe(Observer<R> observableEmitter) {
        this.observableEmitter = observableEmitter;

        // source.subscribe(observableEmitter); // 不应该把下一层Observer交出去 ---》 上一层， 如果交出去了，map没有控制权

        MapObserver<T> mapObserver = new MapObserver(observableEmitter, source, function);

        source.subscribe(mapObserver); // 把我们自己 map MapObserver 交出去了

    }

    class MapObserver<T> implements  Observer<T>{
        ObservableOnSubscribe source;

        Function<T,R> function;

        Observer<R>  observableEmitter;

        public MapObserver(Observer<R>  observableEmitter, ObservableOnSubscribe source, Function<T, R> function) {
            this.observableEmitter = observableEmitter;
            this.function =function;
            this.source=source;
        }

        @Override
        public void onSubscribe() {
            observableEmitter.onSubscribe();
        }

        @Override
        public void onNext(T item) {
            try {
               R r=  function.apply(item);

                observableEmitter.onNext(r);

            } catch (Throwable throwable) {
                throwable.printStackTrace();
                observableEmitter.onError(throwable);

            }
        }

        @Override
        public void onError(Throwable e) {
            observableEmitter.onError(e);
        }

        @Override
        public void onComplete() {
            observableEmitter.onComplete();
        }
    }
}
